package com.and.video.service.config;


import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.and.video.domain.UserFollowing;
import com.and.video.domain.UserMoment;
import com.and.video.domain.constant.UserMomentsConstant;
import com.and.video.service.UserFollowingService;
import com.and.video.service.websocket.WebSocketService;
import io.netty.util.internal.StringUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Configuration
public class RocketMQConfig {

    @Value("${rocketmq.name.server.address}")
    private String nameServerAddr;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private UserFollowingService userFollowingService;

    // 生产者
    @Bean("momentsProducer")
    public DefaultMQProducer momentsProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(UserMomentsConstant.GROUP_MOMENTS);
        producer.setNamesrvAddr(nameServerAddr);
        producer.start();
        return producer;
    }

    /**
     * 消费新的动态
     * 发给每个粉丝
     * 每个粉丝获取自己在redis中的关注的人发出的动态的列表
     * 添加新的关注者的动态
     * 放入redis
     */
    @Bean("momentsConsumer")
    public DefaultMQPushConsumer momentsConsumer() throws Exception{
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentsConstant.GROUP_MOMENTS);
        consumer.setNamesrvAddr(nameServerAddr);
        // 设置订阅主题
        // 通配符表示订阅该 Topic 下的所有 Tag
        consumer.subscribe(UserMomentsConstant.TOPIC_MOMENTS, "*");
        // 注册并发消息监听器
        consumer.registerMessageListener((MessageListenerConcurrently) (
                // 消息
                msgs,
                // 上下文
                context) -> {
            // 发布一个动态, 获取一个进行处理
            MessageExt msg = msgs.get(0);
            if(msg == null){
                // 消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            // 生产者发送的是二进制格式化的数据
            // 转换成string
            String bodyStr = new String(msg.getBody());
            // 转换成userMoment对象
            UserMoment userMoment = JSONObject.toJavaObject(JSONObject.parseObject(bodyStr), UserMoment.class);
            // 获取用户id
            Long userId = userMoment.getUserId();
            // 获取用户粉丝, 把自己的动态发给自己的粉丝
            List<UserFollowing>fanList = userFollowingService.getUserFans(userId);
            for(UserFollowing fan : fanList){
                String key = "subscribed-" + fan.getUserId();
                // 从redis获取字符串
                // subscribed-11 发给用户id:11的动态
                String subscribedListStr = redisTemplate.opsForValue().get(key);
                List<UserMoment> subscribedList;
                if(StringUtil.isNullOrEmpty(subscribedListStr)){
                    // 字符串为空,生成空列表
                    subscribedList = new ArrayList<>();
                }else{
                    // 字符串不为空,转换成UserMoment列表
                    subscribedList = JSONArray.parseArray(subscribedListStr, UserMoment.class);
                }
                // 添加进去新的动态
                subscribedList.add(userMoment);
                redisTemplate.opsForValue().set(key, JSONObject.toJSONString(subscribedList));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 启动并返回
        consumer.start();
        return consumer;
    }

    // 创建RocketMQ生产者Bean，用于发送弹幕消息
    @Bean("danmusProducer")
    public DefaultMQProducer danmusProducer() throws Exception{
        // 实例化消息生产者Producer，指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer(UserMomentsConstant.GROUP_DANMUS);
        // 设置NameServer的地址，用于服务发现和路由管理
        producer.setNamesrvAddr(nameServerAddr);
        // 启动Producer实例
        producer.start();
        return producer;
    }

    // 创建RocketMQ消费者Bean，用于接收和处理弹幕消息
    @Bean("danmusConsumer")
    public DefaultMQPushConsumer danmusConsumer() throws Exception{
        // 实例化消费者，指定消费者组名（与生产者组名相同）
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentsConstant.GROUP_DANMUS);
        // 设置NameServer的地址
        consumer.setNamesrvAddr(nameServerAddr);
        // 订阅弹幕主题，使用"*"表示接收所有标签的消息
        consumer.subscribe(UserMomentsConstant.TOPIC_DANMUS, "*");
        // 注册消息监听器，处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 获取消息（通常列表里只有一条消息）
                MessageExt msg = msgs.get(0);
                // 获取消息体字节数组
                byte[] msgByte = msg.getBody();
                // 将字节数组转换为字符串
                String bodyStr = new String(msgByte);
                // 解析JSON格式的消息体
                JSONObject jsonObject = JSONObject.parseObject(bodyStr);
                // 从JSON中提取sessionId和消息内容
                String sessionId = jsonObject.getString("sessionId");
                String message = jsonObject.getString("message");

                // 根据sessionId从WebSocket连接映射中获取对应的WebSocket服务实例
                WebSocketService webSocketService = WebSocketService.WEBSOCKET_MAP.get(sessionId);

                // 检查WebSocket连接是否仍然打开
                if(webSocketService != null && webSocketService.getSession().isOpen()){
                    try {
                        // 通过WebSocket连接向客户端发送消息
                        webSocketService.sendMessage(message);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        return consumer;
    }

}